/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.protocol;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Enumeration;
import java.util.Hashtable;

import cgl.narada.event.DistributionTraces;
import cgl.narada.event.EventHeaders;
import cgl.narada.event.NBEvent;
import cgl.narada.service.security.kmc.SignedSecurityToken;
import cgl.narada.service.timer.ClockI;
import cgl.narada.service.timer.HRClock;
import cgl.narada.util.ByteUtilities;

/**
 * The EventRoutingProtocol is responsible for updating routing information
 * associated with events. The routing information would vary depending on the
 * link over which the event was received. Any given node could be connected to
 * multiple links which have different gateway behavior. The event routing
 * protocol needs to handle every such case.
 * 
 * @author Shrideep Pallickara
 * @author Harshawardhan Gadgil $Revision$ $Date$
 */

public class EventRoutingProtocol extends Thread implements ProtocolDebugFlags {
    /** The node address of this node */
    private byte[] nodeAddress;

    /** The system level of the system to which this node is attached to */
    private int systemLevel;

    /** Maintains information about revelant gatekeepers within the system */
    private GatewayInfo gatewayInfo;

    /* Reference to the profile propagation protocol */
    ProfilePropagationProtocol ppProtocol;

    /** A handle to the protocol Handler */
    private ProtocolHandler protocolHandler;

    /**
     * Used for routing events to clients after the destinations are computed by
     * a matchingtree at level-0
     */
    private ClientConnectionHandler clientConnectionHandler;

    /** The destination ID of this node */
    private Destinations thisDestination;

    private DuplicateDetection duplicateDetection;

    private String moduleName = "EventRoutingProtocol:";

    private long timeLastProcessed;

    // HG: Benchmark..
    private boolean benchmark = false;

    private FileOutputStream fout = null;

    private ClockI hrClock = null;

    private EventBuffer eventBuffer;


    /**
     * The constructor requires both the nodeAdress which it would use to update
     * routing information associated with a certain received event, and also
     * the gatewayInfo which would be used to compute the next most efficient
     * hop
     * 
     * @param _nodeAddress
     *            The node address of the node in question.
     * @param gatewayInfo
     *            Reference to the connectivity graph hosted at the node
     * @param ppProtocol
     *            Reference to the profile propagation protocol which would
     *            provide uswith destinations at different levels.
     * @param clientConnectionHandler
     *            This is useful for routing events to clients within the
     *            destination lists that are generated.
     * @param protocolHandler
     *            A handle to the protocol Handler
     */
    public EventRoutingProtocol(NodeAddress _nodeAddress,
            GatewayInfo gatewayInfo, ProfilePropagationProtocol ppProtocol,
            ClientConnectionHandler clientConnectionHandler,
            ProtocolHandler protocolHandler) {
        nodeAddress = _nodeAddress.getAddressInBytes();
        thisDestination = new Destinations(nodeAddress);
        this.gatewayInfo = gatewayInfo;
        this.clientConnectionHandler = clientConnectionHandler;
        this.ppProtocol = ppProtocol;
        this.protocolHandler = protocolHandler;
        systemLevel = gatewayInfo.getSystemLevel();

        duplicateDetection = DuplicateDetection.getInstance();
        // BENCHMARK
        if (benchmark) {
            try {
                fout = new FileOutputStream("/home/hgadgil/ERP.out");
            } catch (FileNotFoundException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                System.err.println("Could not open benchmark output file..."
                    + e);
            }

            hrClock = HRClock.getClock();
        }
        eventBuffer = EventBuffer.getInstance();
        start();
    }


    public void run() {
        while (true) {
            try {
                NBEvent nbEvent = eventBuffer.getEvent();
                processEvent(nbEvent);
            } catch (InterruptedException ie) {
                System.out.println(moduleName + "Problems retrieving event "
                    + ie);
            }
        }
    }


    /**
     * This method is responsible for ensuring the dissemination of events
     * within the system. This event could have been received from a client or
     * from a neighboring broker
     */
    public void processEventReceived(byte[] nbEventBytes) {
        NBEvent nbEvent =
            cgl.narada.event.impl.NBEventGenerator
                .unmarshallEvent(nbEventBytes);
        
        /** Perform duplicate detection related processing */
        boolean isDuplicate = duplicateDetection.isDuplicate(nbEvent);
        if (isDuplicate) {
            return;
            /** This event has been processed before */
        } else {
            duplicateDetection.registerEvent(nbEvent);
        }

        eventBuffer.addEvent(nbEvent);
    }


    private void processEvent(NBEvent nbEvent) {        
        byte[] nbEventBytes = nbEvent.getBytes();
        long start = 0, end = 0;
        // cgl.narada.event.impl.NBEventGenerator.validateIntegrityInformation(nbEvent);

        // HG: Added extra processing for secure events...

        // 1 .Check if the content synopsis is associated with a secure topic
        // 2. Verify that all credentials are correct and then register the
        // synopsis as a secure topic...
        // 3. If everything checks out (Either its not a secure topic OR its a
        // secure and valid event) we set validated to true.
        // 4. If not, Then register this topic synopsis as a secure topic

        // BENCHMARK (ONLY FOR secure events)
        if (benchmark && nbEvent.getEventHeaders().isSecure()) {
            start = hrClock.getTimeMicroseconds();
        }

        boolean validated = protocolHandler.doSecurityChecks(nbEvent);

        // BENCHMARK (ONLY FOR secure events)
        if (benchmark && nbEvent.getEventHeaders().isSecure()) {
            end = hrClock.getTimeMicroseconds();
            try {
                fout.write(("DoSecurityCheck: " + (end - start) + "\n")
                    .getBytes());
                fout.flush();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        // Route ONLY if a secure event is validated

        if (validated) {
            if (nbEvent.hasDistributionTraces()) {
                processEventReceivedFromBroker(nbEvent, nbEventBytes);

            } else {

                // IF its a secure event and it came from a client,
                // then verify that the publish token is still valid
                EventHeaders hdr = nbEvent.getEventHeaders();

                if (hdr.isSecure()) {
                    // Verify the Signed security token's certificate
                    SignedSecurityToken sst =
                        SignedSecurityToken.createObjectFromBytes(nbEvent
                            .getEventHeaders().getSignedSecurityToken());

                    // Check if SST is valid...
                    boolean ok2Go = false;

                    ok2Go =
                        sst.getSecurityToken().getRights().hasPublishRight();
                    ok2Go &= sst.getSecurityToken().isValid();

                    if (!ok2Go) {
                        System.out.println("Dropping Event ! Token Invalid !!");
                        return;
                    }
                }
                if (EventRoutingProtocol_Debug) {
                    System.out.println(moduleName
                        + "Processing event from CLIENT");
                }
                processEventReceivedFromClient(nbEvent, nbEventBytes);
            }
        }

        if (EventRoutingProtocol_Debug) {
            long _delay = System.currentTimeMillis() - timeLastProcessed;
            System.out.println(moduleName + "[" + nbEvent.getContentSynopsis()
                + "] Inter delay = " + _delay);
            timeLastProcessed = System.currentTimeMillis();
        }

        // Still must do GC stuff...
        // ---

        // controlMemoryUtilization();
    }


    /**
     * This method is used to control memory utlization. The method will not
     * return until the memory utilization has actually gone down.
     */
    private void controlMemoryUtilization() {
        long totalMemory = 0;
        long UTIL_LIMIT = 15000000;
        totalMemory = Runtime.getRuntime().totalMemory();

        if (totalMemory > UTIL_LIMIT) {
            System.gc();
        }
    }


    /**
     * This method is responsible for processing events received from a client
     */
    private void processEventReceivedFromClient(NBEvent nbEvent,
                                                byte[] nbEventBytes) {

        /** Now we begin the task of constructing the routing information */
        Destinations toReach = new Destinations(systemLevel);

        Destinations traversedSoFar = new Destinations(systemLevel);
        traversedSoFar.markAsTraversedAt(thisDestination);

        /**
         * If this node is a gateway at different levels we further refine the
         * destinations associated with the event
         */
        for (int i = 0; i <= systemLevel; i++) {
            // :: if (ppProtocol.isGatewayAtLevel(i + 1)) {
            Integer dest =
                (Integer) ppProtocol
                    .computeEventDestinationsAtLevel(nbEvent, i);
            int destValue = dest.intValue();

            if (EventRoutingProtocol_Debug) {
                String report =
                    moduleName + "Destinations at level (" + i + ") = "
                        + ByteUtilities.printInt(destValue);
                System.out.println(report);
            }
            toReach.setDestinationsAtLevel(i, destValue);

            // ::
            /**
             * if (destValue == 0) { // The event shouldn't be disseminated with
             * the level - i for (int j = i; j >= 0; j--) {
             * toReach.setDestinationsAtLevel(j, destValue); } }// end for
             * (dest==0)
             */
            // :: } /* end if(isGateway) */
        } /* end for */
        // ::
        toReach.updateWithDestination(thisDestination);

        if (EventRoutingProtocol_Debug)
            System.out.println(moduleName + "processEventFromClient() "
                + "Server Destinations to be reached ->" + toReach);
        disseminateEventThroughoutSystem(nbEvent, toReach, traversedSoFar);

        /** Distribute the event to attached clients */
        try {
            routeToClientNodes(nbEvent, nbEventBytes);

        } catch (Exception e) {
            e.printStackTrace();
        }
        nbEvent = null;
        nbEventBytes = null;
        // printMemoryUtilization();
    }


    /**
     * This method is responsible for the calculation of destination lists
     * associated with the event. If this node can compute destinations at a
     * certain level it should do so and if it cannot calculate destinations at
     * a certain level, it should defer the calculation of destination lists to
     * the appropriate gateways. The method should be able to deal with both,
     * upward and downward dissemination of events within the system.
     */
    private void processEventReceivedFromBroker(NBEvent nbEvent,
                                                byte[] nbEventBytes) {
        DistributionTraces traces = nbEvent.getDistributionTraces();

        byte[] toReachBytes = traces.getDestinationsToReach();
        byte[] traversedSoFarBytes = traces.getDestinationsTraversedSoFar();

        Destinations toReach = new Destinations(toReachBytes);
        Destinations traversedSoFar = new Destinations(traversedSoFarBytes);

        traversedSoFar.markAsTraversedAt(thisDestination);

        /**
         * Based on toReach we have to decide how exactly we are going to
         * calculate our destinations that need to be reached.
         */
        // System.out.println(moduleName + "Computing destinations ...");
        // ::
        // calculateDestinations(nbEvent, toReach, systemLevel - 1);
        computeDestinations(nbEvent, toReach, traversedSoFar);

        disseminateEventThroughoutSystem(nbEvent, toReach, traversedSoFar);

        /** Distribute the event to attached clients */
        routeToClientNodes(nbEvent, nbEventBytes);

        nbEvent = null;
        nbEventBytes = null;
        // printMemoryUtilization();
    }


    private void routeToClientNodes(NBEvent nbEvent, byte[] nbEventBytes) {

        Hashtable destinations =
            (Hashtable) ppProtocol.computeEventDestinationsAtLevel(nbEvent, -1);

        if (destinations == null) {
            /** There were no matching destinations */
            if (EventRoutingProtocol_Debug) {
                System.out.println(moduleName + ".routeToClientNodes() "
                    + "No destinations associated with "
                    + nbEvent.getContentSynopsis());
            }
            return;
        }

        if (destinations.size() == 0) {
            // if (EventRoutingProtocol_Debug)
            if (EventRoutingProtocol_Debug)
                System.out.println(moduleName + ".routeToClientNodes()"
                    + "-> No clients can be reached ");
            return;
        }

        // HG:If the event is a secure event
        // check all the destinations to see if their profiles contain the
        // appropriate security token
        if (nbEvent.getEventHeaders().isSecure()) {
            for (Enumeration e = destinations.keys(); e.hasMoreElements();) {
                Integer key = (Integer) e.nextElement();
                if (EventRoutingProtocol_Debug)
                    System.out.println("***Checking if [" + key
                        + "] had specified a VALID security token !");

                SignedSecurityToken o =
                    protocolHandler.getSecurityTokenForProfileId(key);
                if (o == null) {
                    // This dude must not recieve the event, so remove him from
                    // the destinations list
                    if (EventRoutingProtocol_Debug)
                        System.out.println("SST is NULL! Removing: " + key);
                    destinations.remove(key);
                } else {
                    // Check if this guy has subscribe rights
                    if (!o.getSecurityToken().getRights().hasSubscribeRight()) {
                        // NOTE: This auto removes all future recieve rights !
                        if (EventRoutingProtocol_Debug)
                            System.out.println("Removing " + key
                                + " from destinations table !");
                        destinations.remove(key);
                    } else {
                        // ELSE This entity can get the event !
                        if (EventRoutingProtocol_Debug)
                            System.out.println("Destination: " + key
                                + " spared !!:)");
                    }
                }
            }
        }
        // ---

        
        if (nbEvent.hasEventHeaders()) {
            if (nbEvent.getEventHeaders().supressDistributionToSource()) {
                int ignoreSource = nbEvent.getEventHeaders().getSource();
                if (EventRoutingProtocol_Debug) {
                    System.out.println(moduleName
                        + "Supressing distribution to source " + ignoreSource);
                }
                clientConnectionHandler.sendToClientNodes(nbEventBytes,
                                                          destinations,
                                                          ignoreSource);
                return;
            }
        }

        clientConnectionHandler.sendToClientNodes(nbEventBytes, destinations);
    }


    /**
     * This method is responsible for routing events throughout the server
     * network
     * 
     * @param nbEvent
     *            The event that needs to be routed.
     * @param toReach
     *            The destinations that need to be reached.
     * @param traversedSoFar
     *            The list of destinations that have been traversed so far.
     */
    private void disseminateEventThroughoutSystem(NBEvent nbEvent,
                                                  Destinations toReach,
                                                  Destinations traversedSoFar) {

        Gateway[] hopsToTake =
            gatewayInfo.hopsToReachDestination(toReach, traversedSoFar);
        // ::
        if (hopsToTake != null) {
            for (int i = 0; i < hopsToTake.length; i++) {
                Gateway _gateway = hopsToTake[i];
                traversedSoFar.updateDestinationList(_gateway);
            }
        } else {
            if (EventRoutingProtocol_Debug) {
                System.out.println(moduleName + "Hops to take is NULL!");
            }
        }
        // ::

        if (EventRoutingProtocol_Debug) {
            String information =
                moduleName + "To Reach: \n" + toReach.toString()
                    + "Traversed so far: \n" + traversedSoFar.toString();
            System.out.println(information);
        }

        /** if connection.level < hop.level ... don not send over the hop */
        if (hopsToTake != null) {
            for (int i = 0; i < hopsToTake.length; i++) {

                /**
                 * Depending on the kind of hop that you are gonna be sending
                 * the connection ... send the relevant connection information
                 * over the link
                 */

                byte[] byteStream =
                    prepareEventBytesToSendOverHop(hopsToTake[i].getLevel(),
                                                   nbEvent, toReach,
                                                   traversedSoFar);
                protocolHandler.sendToNode(hopsToTake[i], byteStream);

                if (EventRoutingProtocol_Debug) {
                    System.out.println(moduleName
                        + "Byte stream sent over hop " + hopsToTake[i]);
                }
            }
            return;
        }
        if (EventRoutingProtocol_Debug)
            System.out.println(moduleName + "No more hops to take");
    }


    /**
     * This method is responsible for sending information over a hop. The
     * information contained in the Destinations are modified based on the type
     * of hop that we are going to be sending this event over
     */
    private byte[] prepareEventBytesToSendOverHop(int hopLevel,
                                                  NBEvent nbEvent,
                                                  Destinations toReach,
                                                  Destinations traversedSoFar) {
        DistributionTraces traces = nbEvent.getDistributionTraces();

        byte[] toReachBytes =
            toReach.prepareDestinationsToSendOverLevel(hopLevel);
        byte[] traversedSoFarBytes =
            traversedSoFar.prepareDestinationsToSendOverLevel(hopLevel);

        traces.setDestinationsToReach(toReachBytes);
        traces.setDestinationsTraversedSoFar(traversedSoFarBytes);
        cgl.narada.event.impl.NBEventGenerator
            .updateDistributionTraces(nbEvent, traces);
        return nbEvent.getBytes();
    }


    private void computeDestinations(NBEvent nbEvent, Destinations toReach,
                                     Destinations traversedSoFar) {

        int level = 0;
        int destAtLevel = 0;

        while (destAtLevel == 0) {
            destAtLevel = toReach.getDestinationsAtLevel(level);

            if (destAtLevel == 0) {
                level++;
            }
        }

        if (EventRoutingProtocol_Debug) {
            System.out.println(moduleName + "Event received over a level ("
                + level + ") link");
        }

        for (int i = level - 1; i >= 0; i--) {
            if (EventRoutingProtocol_Debug) {
                System.out.println(moduleName
                    + "Computing destinations at level [" + i + "]");
            }
            Integer dest =
                (Integer) ppProtocol
                    .computeEventDestinationsAtLevel(nbEvent, i);
            int destValue = dest.intValue();

            toReach.setDestinationsAtLevel(i, destValue);
        }

    }


    /**
     * This method recursively computes destinations for lower levels. The
     * recursion stops if the node in question is not a gateway at one of the
     * levels (starting from highest levels to succesively lower ones)
     * 
     * @param nbEvent
     *            The event for which destinations need to be computed
     * @param toReach
     *            The destinations that need to be reached
     */
    public void calculateDestinations(NBEvent nbEvent, Destinations toReach,
                                      int computeForLevel) {
        // System.out.println("Computing destinations at " + computeForLevel);
        int _level = computeForLevel;
        if (_level == -1) {
            /**
             * No server destinations need to be computed, we compute the client
             * destinations and move on
             */
            return;
        }

        for (; _level == 0; _level--) {
            if (toReach.getDestinationsAtLevel(_level) == 0) {
                /**
                 * This implies we need to compute destinations from level i
                 * onwards down to level 0
                 */
                break;
            }
        }

        /** Only if its a gateway at (l+1) can it compute destinations at level-l */
        if (ppProtocol.isGatewayAtLevel(_level + 1)) {
            Integer dest =
                (Integer) ppProtocol.computeEventDestinationsAtLevel(nbEvent,
                                                                     _level);
            int destValue = dest.intValue();

            if (EventRoutingProtocol_Debug)
                System.out.println(moduleName + "Gateway at level ("
                    + (_level + 1) + ") computing destinations at " + _level
                    + ByteUtilities.printInt(destValue));
            /**
             * This statement prevents a node from being stuck in a infinite
             * recursive process in the event that dest=0
             */
            if (dest.intValue() == 0)
                return;
            toReach.setDestinationsAtLevel(_level, destValue);
            calculateDestinations(nbEvent, toReach, (_level - 1));
        }
    }


    public static void main(String[] args) {

    }
}
